
public void onMessage(Message message, byte[] pattern) {
byte[] messageChannel = message.getChannel();
byte[] messageBody = message.getBody();
if (messageChannel == null || messageBody == null) {
    return;
}
try {
    Command cmd = Command.parse(String.valueOf(SerializationUtils.deserialize(messageBody)));
    if (cmd == null || cmd.isLocal())
        return;

    switch (cmd.getOperator()) {
        case Command.OPT_JOIN:
            logger.info("Node-"+cmd.getSrc()+" joined to " + this.channel);
            break;
        case Command.OPT_EVICT_KEY:
            clusterPolicy.evict(cmd.getRegion(), cmd.getKeys());
            logger.debug("Received cache evict message, region=" + cmd.getRegion() + ",key=" + String.join(",", cmd.getKeys()));
            break;
        case Command.OPT_CLEAR_KEY:
            clusterPolicy.clear(cmd.getRegion());
            logger.debug("Received cache clear message, region=" + cmd.getRegion());
            break;
        case Command.OPT_QUIT:
            logger.info("Node-"+cmd.getSrc()+" quit to " + this.channel);
            break;
        default:
            logger.warn("Unknown message type = " + cmd.getOperator());
    }
} catch (Exception e) {
    logger.error("Failed to handle received msg", e);
}
}